package com.kafka.consumer;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.ProductCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.web.client.HttpServerErrorException;
import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestTemplate;

import java.util.List;

/**
 * @author Administrator
 */
//@Component

public class ConsumerListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerListener.class);
    private RestTemplate restTemplate;

    public ConsumerListener(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }
    @KafkaListener(id = "one",clientIdPrefix = "one",topics = "huran-test", concurrency = "1",groupId = "ggg")
    public void handleGroup1(@Payload ProductCreatedEvent record, Acknowledgment acknowledgment,
                             @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) String key,
                             @Header(name = KafkaHeaders.RECEIVED_TOPIC) String topic,
                             @Header(name = KafkaHeaders.RECEIVED_PARTITION) String partition,
                             @Header(name = KafkaHeaders.RECEIVED_TIMESTAMP) String ts) {
        LOGGER.info("========================阻塞7秒========================");
        try {
            Thread.sleep(7000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        LOGGER.info("========================handleGroup1========================");
        LOGGER.info(Thread.currentThread().getName() + "+收到消息:" + record.getTitle());
        LOGGER.info(Thread.currentThread().getName() + "+topic:" + topic);
        LOGGER.info(Thread.currentThread().getName() + "+partition:" + partition);
        LOGGER.info(Thread.currentThread().getName() + "+ts:" + ts);
        LOGGER.info(Thread.currentThread().getName() + "+key:" + key);
        LOGGER.info("========================handleGroup1========================");

        acknowledgment.acknowledge();
    }
//    @KafkaListener(id = "two",clientIdPrefix = "two",topics = "huran-test", concurrency = "1",groupId = "ggg")
//    public void handleGroup2(ProductCreatedEvent record, Acknowledgment acknowledgment,
//                       @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) String key,
//                       @Header(name = KafkaHeaders.RECEIVED_TOPIC) String topic,
//                       @Header(name = KafkaHeaders.RECEIVED_PARTITION) String partition,
//                       @Header(name = KafkaHeaders.RECEIVED_TIMESTAMP) String ts) {
//        LOGGER.info("========================handleGroup2========================");
//        LOGGER.info(Thread.currentThread().getName() + "+收到消息:" + record.getTitle());
//        LOGGER.info(Thread.currentThread().getName() + "+topic:" + topic);
//        LOGGER.info(Thread.currentThread().getName() + "+partition:" + partition);
//        LOGGER.info(Thread.currentThread().getName() + "+ts:" + ts);
//        LOGGER.info(Thread.currentThread().getName() + "+key:" + key);
//        LOGGER.info("========================handleGroup2========================");
//
//        acknowledgment.acknowledge();
//    }
    /**
     * 演示消费失败重试
     */
//    @KafkaListener(topics = "huran-test", concurrency = "3")
//    public void handle(ProductCreatedEvent record, Acknowledgment acknowledgment,
//                       @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) String key,
//                       @Header(name = KafkaHeaders.RECEIVED_TOPIC) String topic,
//                       @Header(name = KafkaHeaders.RECEIVED_PARTITION) String partition,
//                       @Header(name = KafkaHeaders.RECEIVED_TIMESTAMP) String ts) {
//        LOGGER.info("=========================================================");
//        LOGGER.info(Thread.currentThread().getName() + "+收到消息:" + record.getTitle());
//        LOGGER.info(Thread.currentThread().getName() + "+topic:" + topic);
//        LOGGER.info(Thread.currentThread().getName() + "+partition:" + partition);
//        LOGGER.info(Thread.currentThread().getName() + "+ts:" + ts);
//        LOGGER.info(Thread.currentThread().getName() + "+key:" + key);
//        LOGGER.info("=========================================================");
//
//        acknowledgment.acknowledge();
//        String url = "http://localhost:8082";
//        try {
//            ResponseEntity<String> response = restTemplate.exchange(url, HttpMethod.GET, null, String.class);
//            if (response.getStatusCode().value() == HttpStatus.OK.value()) {
//                LOGGER.info("远程小消息响应:" + response.getBody());
//            }
//        } catch (ResourceAccessException e) {
//            System.out.println("===============>>>ResourceAccessException");
//            // acknowledgment.acknowledge();
//            LOGGER.error(e.getMessage());
//            throw new RetryableException(e);
//        } catch (HttpServerErrorException e) {
//            System.out.println("===============>>>HttpServerErrorException");
//            acknowledgment.acknowledge();
//            LOGGER.error(e.getMessage());
//            throw new NotRetryableException(e);
//        } catch (Exception e) {
//            LOGGER.error(e.getMessage());
//            throw new NotRetryableException(e);
//        }

   // }


//    /**
//     * 演示消费失败 不重试 发送到私信队列
//     * @param record
//     */
//    @KafkaListener(topics = "huran-test")
//    public void handle(ProductCreatedEvent record) {
//        if (true) {
//            throw new NotRetryableException("不重试异常，放到死信队列");
//        }
//        LOGGER.info("收到方法:{}", record.getTitle());
//    }


//    @KafkaListener(topics = "huran-test")
//    public void handle(ProductCreatedEvent record, Acknowledgment acknowledgment) {
//
//        LOGGER.info("收到方法:{}", record.getTitle());
//        acknowledgment.acknowledge();
//    }
}

//    @KafkaListener(
//            id = "myConsumerId",
//            containerFactory = "kafkaListenerContainerFactory",
//            topics = "huran-test",
//            groupId = "your-group-id",
//            clientIdPrefix = "myClientPrefix",
//            concurrency = "3",
//            autoStartup = "true"
//    )
//    public void handle(List<ConsumerRecord<String, ProductCreatedEvent>> records, Acknowledgment acknowledgment) {
//        try {
//            for (ConsumerRecord<String, ProductCreatedEvent> record : records) {
//                String topic = record.topic();
//                int partition = record.partition();
//                long offset = record.offset();
//                ProductCreatedEvent event = record.value();
//                System.out.println(event);
//
//            }
//
//            // 批量处理成功后手动确认
//            acknowledgment.acknowledge();
//        } catch (Exception e) {
//            LOGGER.error("处理消息失败: {}", e.getMessage(), e);
//            // 处理失败时可以选择不确认消息，让Kafka重新发送
//        }
//    }
//}
